package com.study.connect.client.handler;

import com.study.connect.client.impl.ConnectServiceImpl;
import com.study.connect.exception.MyRpcConnectionCloseException;
import com.study.connect.exception.MyRpcTimeOutException;
import com.study.connect.message.ServiceInvokeDTO;
import io.netty.channel.ChannelFuture;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @包名称 com.study.connect.client.handler
 * @类描述 <pre></pre>
 * @作者 zhangbo
 * @创建时间 2020/6/13.
 * @版本 1.00
 * @修改记录 <pre>
 *     版本           修改人         修改日期         修改内容描述
 *     -----------------------------------------------------------
 *     1.00           zhangbo        2020/6/13.           类创建
 *     -----------------------------------------------------------
 * </pre>
 */
@Slf4j
@Data
public class HandlerCallable implements Callable {

    private volatile Object result;

    private ChannelFuture channelFuture;

    private ServiceInvokeDTO serviceInvokeDTO;

    private CountDownLatch countDownLatch;

    private int timeout;

    public HandlerCallable(ChannelFuture channelFuture, ServiceInvokeDTO serviceInvokeDTO, int timeout) {
        this.channelFuture = channelFuture;
        this.serviceInvokeDTO = serviceInvokeDTO;
        ConnectServiceImpl.CLIENT_HANDLER_LOCAL.get().setHandlerCallable(this);
        countDownLatch = new CountDownLatch(1);
        this.timeout = timeout;
    }

    @Override
    public Object call() throws MyRpcTimeOutException, InterruptedException {
        if (!channelFuture.channel().isActive()) {
            throw new MyRpcConnectionCloseException("连接已关闭");
        }
        channelFuture.channel().writeAndFlush(serviceInvokeDTO);
        boolean await = countDownLatch.await(timeout, TimeUnit.MILLISECONDS);
        if (!await) {
            throw new MyRpcTimeOutException("服务调用超时");
        }
        log.debug("返回参数：{}", result);
        return result;
    }


}
